package com.demo.userprofile.component.service.democode;

import java.util.Properties;

import org.apache.commons.lang3.time.DateFormatUtils;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer09;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
import org.apache.flink.util.Collector;
import org.springframework.beans.factory.annotation.Autowired;

import com.demo.userprofile.component.service.RedisService;
import com.google.gson.JsonObject;
import com.google.gson.JsonParser;

import lombok.AllArgsConstructor;
import lombok.Data;

/**
 * Flink消费Kafka示例代码
 *
 * @author userprofile_demo
 */
public class FlinkConsumeKafkaData {

    @Autowired
    private static RedisService redisService;

    @Data
    @AllArgsConstructor
    public static class Event {
        private Long userId;
        private Long photoId;
        private long shareTime;
    }

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env =
                StreamExecutionEnvironment.getExecutionEnvironment();

        // Kafka配置
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "Kafka IP及端口号");
        properties.setProperty("group.id", "消费Group");
        properties.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        properties.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        properties.setProperty("auto.offset.reset", "latest");

        // 实时数据流配置
        DataStreamSource<String> stream = env.addSource(new FlinkKafkaConsumer09<String>(
                "Topic名称",
                new SimpleStringSchema(),
                properties
        ));

        // 解析数据流中的每一行数据，从json中获取关键属性信息
        SingleOutputStreamOperator<Event> flatMap = stream.flatMap(new FlatMapFunction<String, Event>() {
            @Override
            public void flatMap(String message, Collector<Event> collector) throws Exception {
                // 解析message中的数据
                JsonObject jsonObject = JsonParser.parseString(message).getAsJsonObject();
                long userId = jsonObject.get("userId").getAsLong();
                long photoId = jsonObject.get("photoId").getAsLong();
                long shareTime = jsonObject.get("shareTime").getAsLong();
                collector.collect(new Event(userId, photoId, shareTime));
            }
        });

        // Flink连接Redis配置
        FlinkJedisPoolConfig config = new FlinkJedisPoolConfig.Builder()
                .setHost("Redis IP地址")
                .setDatabase(15)
                .build();

        flatMap.addSink(new RichSinkFunction<Event>() {
            @Override
            public void invoke(Event value, Context context) throws Exception {
                // 根据shareTime获取当前日期20220626
                String currentDate = DateFormatUtils.format(value.getShareTime(), "yyyyMMdd");
                // 构建Redis Key，示例：dt:20220626:uid:100:like:count
                String redisKey = String.format("dt:%s:uid:%s:share:count", currentDate, value.getUserId());
                // 累加当前key的value
                redisService.incrBy(redisKey, 1L);
            }
        });

        env.execute();
    }
}
